package com.shux.kafka.consumer;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.kafka.common.serialization.Deserializer;

/**
 * 描述：
 * 
 * <pre>HISTORY
 * ****************************************************************************
 *  ID   DATE           PERSON          REASON
 *  1    2017年9月22日      Simba.Hua         Create
 * ****************************************************************************
 * </pre>
 * @author Simba.Hua
 */
public class TransactionalKafkaConsumer<T> extends ShuxKafkaConsumer<T> {
	private Map<Integer,OffsetStatusHolder> partitionOffsetStatus = new HashMap<>();
	
	public TransactionalKafkaConsumer(ConsumerConfig config,
			ConsumerOptionConfig optionConfig,IMessageListener<T> messageListener,Deserializer<T> deserializer) {
		super(config, optionConfig,messageListener,deserializer);
	}
	/**
	 * 逻辑
	 * 迭代KafkaBatchMessage的每个partition的Offset
	 * 
	 */
	@Override
	protected void beforeAddPackMessageToQueue(KafkaBatchMessage<T> batchMessage) {
		Map<Integer,Long> partitionOffset = batchMessage.getOffsets();
		Set<Integer> removePartition = new HashSet<>();
		partitionOffset.forEach((partition,offset) -> {
			OffsetStatusHolder offsetStatusHolder = partitionOffsetStatus.get(partition);
			if(offsetStatusHolder == null) {
				partitionOffsetStatus.put(partition, new OffsetStatusHolder());
			} 
			if(offsetStatusHolder != null && offsetStatusHolder.checkIsConsumered(offset)) {
				removePartition.add(partition);
			}
		});
		if (removePartition.size() > 0) {
			removePartition.forEach(p -> {
				batchMessage.removeKafkaMessage(p);//把已经消费的移除掉，以免再次消费
			});
		}
	}
	/**
	 * 标记是否已经发送完成
	 */
	@Override
	protected void completeMessage(KafkaBatchMessage<T> batchMessage){
		Map<Integer,Long> partitionOffset = batchMessage.getOffsets();
		partitionOffset.forEach((partition,offset) -> {
			OffsetStatusHolder offsetStatusHolder = partitionOffsetStatus.get(partition);
			if(offsetStatusHolder == null) {
				throw new RuntimeException("invaild partition");
			}
			offsetStatusHolder.complete(offset);
		});
	}

	protected void commitOffsetToKafka(){
		CommitOffsetThread<T> commitOffsetThread = new CommitOffsetThread<>(null,this,partitionOffsetStatus);
		commitOffsetThread.start();
		
	}
}
